package com.chwl.framework.grpc.etcd;

import com.alibaba.fastjson.JSON;
import com.chwl.framework.grpc.exception.UserdefinedException;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.watch.WatchEvent;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


/**
 * 包名：com.mechat.connector.etcd<br>
 * <p>功能:etcd封装</p>
 *
 * @date:2018-07-12 下午6:56<br/>
 * @version:1.0 <br/>
 */
public class EtcdService extends EtcdCrudService {
    private Logger logger = LoggerFactory.getLogger(EtcdService.class);
    private static EtcdService instance = new EtcdService();
    ConcurrentHashMap<String, String> localCache = new ConcurrentHashMap<String, String>();

    public EtcdService() {
    }

    public static EtcdService getInstance() {
        return instance;
    }

    /**
     * 打开etcd连接
     *
     * @param endpoints
     * @throws UserdefinedException
     */
    public void connection(String... endpoints) throws UserdefinedException {
        logger.info("etcd service,connection endpoints:{}", JSON.toJSONString(endpoints));
        super.connection(endpoints);
        watch(EtcdBusienssServerKey.ROOT_KEY);
    }

    /**
     * 获取单个
     *
     * @param key
     * @return
     * @throws UserdefinedException
     */
    public String get(String key) throws UserdefinedException {
        String value = localCache.get(key);
        if (StringUtils.isNotEmpty(value)) {
            if (logger.isDebugEnabled()) {
                logger.debug("etcd client,etcd service,get data by cache,key:{},value:{}", key, value);
            }
            return value;
        }
        value = super.get(key);
        if (logger.isDebugEnabled()) {
            logger.debug("etcd client,etcd service,get data by etcd,key:{},value:{}", key, value);
        }
        return value;
    }

    /**
     * watch监听
     *
     * @param key
     * @throws Exception
     */
    public void watch(String key) {
        logger.info("etcd client,etcd service watch data,watch root key start");
        initAll();
        executorService.execute(() -> {
            watch(key, (List<WatchEvent> watchEvents) -> {
                for (WatchEvent event : watchEvents) {
                    KeyValue keyValue = event.getKeyValue();
                    String eventKey = keyValue.getKey().toString(utf8);
                    String eventValue = keyValue.getValue().toString(utf8);
                    String eventType = event.getEventType().toString();
                    if (StringUtils.startsWith(eventKey, EtcdBusienssServerKey.RPC_SERVICE)) {
                        continue;
                    }
                    heartbeatLogger.info("etcd client,etcd service watch data,event:{},key:{},value:{}", eventType, eventKey, eventValue);
                    if (eventType.equals(WatchEvent.EventType.PUT.toString())) {
                        localCache.put(eventKey, keyValue.getValue().toString(utf8));
                        continue;
                    }
                    if (eventType.equals(WatchEvent.EventType.DELETE.toString())) {
                        localCache.remove(keyValue.getKey().toString(utf8));
                        continue;
                    }
                }
            });
        });
    }

    private void initAll() {
        try {
            List<KeyValue> keyValues = super.findList(EtcdBusienssServerKey.ROOT_KEY);
            for (int i = 0; i < keyValues.size(); i++) {
                KeyValue keyValue = keyValues.get(i);
                String key = keyValue.getKey().toString(utf8);
                String value = keyValue.getValue().toString(utf8);
                localCache.put(key, value);
            }
        } catch (Exception e) {
            logger.error("etcd client,init all data fail,error message:{}", e.getMessage(), e);
        }
    }
}